package action

import (
	"gitee.com/sansaniot/ssiot-core/facade/env"
	"gitee.com/sansaniot/ssiot-core/logger"
	"github.com/tidwall/sjson"
	"gorm.io/gorm"
	"strings"

	"ssdevice/common/constant"
	cDto "ssdevice/common/dto"
	"ssdevice/common/enums"
	"ssdevice/common/middleware/mqtt"
	"ssdevice/common/utils"
	"ssdevice/internal/device/models"
)

type Sensor struct {
	Orm *gorm.DB
}

func (action *Sensor) Execute(topic string, jsonData map[string]interface{}) {
	devSn := strings.Split(topic, "/")[3]
	_, to := AutoLogin(action.Orm, devSn, jsonData)
	if strings.HasSuffix(topic, "/sensor") {
		// 变量上报 - 主题去掉了post
		targetSn := utils.If(len(to) > 0 && to != devSn, to, devSn).(string)
		sensor := utils.GetFieldFromJson([]string{"data"}, jsonData)
		now := utils.GetNowTimeStr()
		// 更新实时数据
		if sensor != nil {
			mqtt.DealData(targetSn, sensor.(map[string]interface{}), func(mData map[string]interface{}) {
				action.SensorUpdate(targetSn, mData, now)
			}, enums.SensorMsg)
		}
		// 更新dev状态
		UpdateLiveStatus(action.Orm, devSn, to, now)
	}
}

// SensorUpdate 测量量sensor更新
func (action *Sensor) SensorUpdate(devSn string, sensor map[string]interface{}, now string) {
	addSensorQueue(devSn, sensor)
	addVarQueue(devSn, sensor, now)
	addHistoryQueue(devSn, sensor, enums.SensorMsg, now)
}

func addSensorQueue(sn string, data map[string]interface{}) {
	var tmpSensor string
	if value, ok := constant.SensorCache.Load(sn); !ok {
		tmpSensor = constant.DevSensorTemplate
	} else {
		tmpSensor = value.(string)
	}
	tmpMap := utils.BytesToMap([]byte(tmpSensor))
	if s, ok := tmpMap["sensor"]; !ok || s == nil {
		tmpMap["sensor"] = make(map[string]interface{})
	}
	// 合并
	result := tmpMap["sensor"].(map[string]interface{})
	utils.DeepMerge(data, result)
	sensor, _ := sjson.Set(tmpSensor, "sensor", result)
	// 再缓存
	constant.SensorCache.Store(sn, sensor)
	// 队列
	constant.SensorQueue.Enqueue(models.SysDevSensor{DevSn: sn, Data: []byte(sensor)})
}

func addVarQueue(sn string, data map[string]interface{}, now string) {
	// 变量表
	if len(data) == 0 {
		return
	}
	flatten := map[string]interface{}{}
	utils.FlattenJson(data, flatten, []string{})
	nowTime := cDto.LocalTime(*utils.FormatStrTimePoint(now))
	vars := make([]models.SysDevVariable, 0)
	for k, v := range flatten {
		vars = append(vars, models.SysDevVariable{
			Sn:         sn,
			Key:        k,
			Value:      utils.InterfaceToStr(v),
			UpdateTime: &nowTime,
		})
	}
	// 多线程消费
	varConsumer := constant.VariableQueue[utils.HashVal(sn)%constant.DataRoutineChild]
	if varConsumer == nil {
		return
	}
	varConsumer.Enqueue(vars)
}

// 加入历史队列
func addHistoryQueue(devSn string, data map[string]interface{}, msgType enums.MsgType, now string) {
	cacheData := map[string]interface{}{
		"sn":         devSn,
		"data":       data,
		"updateTime": now,
		"msgType":    msgType.Get().Name,
	}
	// 视频ai事件上报
	alarmEvent := utils.GetFieldFromJson([]string{"alarm", "event"}, data)
	if alarmEvent != nil {
		dptId, _ := env.Cache.HashGet(constant.DeviceDeptCache, devSn)
		cacheData["deptId"] = dptId
	}
	dataBytes := utils.MapToBytes(cacheData)
	// 多线程消费
	hisConsumer := constant.HisDataBucket[utils.HashVal(devSn)%constant.DataRoutineChild]
	if hisConsumer == nil {
		return
	}
	hisConsumer.Enqueue(string(dataBytes))
	// 广播已处理数据
	mqtt.MqttClient.Publish(constant.HandledMsgTopic, dataBytes)
}

// UpdateLiveStatus 更新活跃状态
func UpdateLiveStatus(db *gorm.DB, devSn, to, now string) {
	dev := &models.SysDev{}
	dev.Model.Orm = db
	if err := dev.UpdateOnlineStatus(devSn, to, now, enums.Online); err != nil {
		logger.Error(err)
	}
}
